package com.smsc.headend.connector.manager;

import cn.hutool.core.convert.Convert;
import com.smsc.headend.common.thread.UDISThreadFactory;
import com.smsc.headend.common.utils.RedisKeys;
import com.smsc.headend.common.utils.RedisUtils;
import com.smsc.headend.connector.bo.ConnectTaskResultDto;
import com.smsc.headend.connector.enums.states.ChannelState;
import com.smsc.headend.connector.netty.NettyConfigHelper;
import com.smsc.headend.connector.netty.initializer.ChannelInitializerFactory;
import com.smsc.headend.connector.service.DeviceConnectService;
import com.smsc.headend.connector.service.TerminalChannelService;
import com.smsc.headend.connector.utils.NetUtil;
import com.smsc.headend.module.asset.entity.ComEntity;
import com.smsc.headend.module.protocol.ProtocolType;
import com.smsc.headend.module.task.dto.ConnectorTask;
import com.smsc.headend.module.task.dto.ConnectorTaskResult;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import java.net.Inet4Address;
import java.net.UnknownHostException;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.*;
import java.util.stream.Collectors;

import static com.smsc.headend.common.enums.errorcode.TaskExecuteErrorCode.CONNECTION_TIMED_OUT;

@Component
@Slf4j
public class TCPClientManager {

    @Autowired
    RedisUtils redisUtils;

    @Autowired
    DeviceConnectService deviceConnectService;

    Map<Long, Bootstrap> bootstrapMap = new HashMap<>();

    NioEventLoopGroup clientEventLoop = new NioEventLoopGroup(Runtime.getRuntime().availableProcessors());

    ScheduledExecutorService registerService = Executors.newScheduledThreadPool(1, new UDISThreadFactory("register"));

    @PostConstruct
    public void initGroupCheck() {
        //启动两分钟后 开始把Registered状态的channel 注册到UDIS中
        registerService.scheduleAtFixedRate(this::registerLimitedChannel, 120, 5, TimeUnit.SECONDS);
    }

    private void registerLimitedChannel() {
        List<Channel> collect = NettyConfigHelper.group.stream().filter(channel -> {
            ChannelState channelState = channel.attr(NettyConfigHelper.channelState).get();
            return ChannelState.WaitingReg.equals(channelState);
        }).collect(Collectors.toList());

        int channelSize = NettyConfigHelper.group.size();
        if (collect.size() > 0) {
            for (Channel channel : collect) {
                deviceConnectService.lazySignIn(channel);
            }
            log.info("currentChannel: totalSize:{}, waitRegisterSize:{}", channelSize, collect.size());
        }
    }

    public void connectAndSendTask(ComEntity com, Long protocolId, ConnectorTask connectorTask) {
        if (!this.bootstrapMap.containsKey(protocolId)) {
            initProtocolBootstrap(protocolId);
        }
        Bootstrap bootstrap = this.bootstrapMap.get(protocolId);
        String netAddr = com.getIpAddress();
        String[] netAddrParts = netAddr.split(":");
        ChannelFuture connect = bootstrap.connect(netAddrParts[0], Convert.toInt(netAddrParts[1]));
        connect.addListener((future -> {
            if (future.isSuccess()) {
                log.info("ConnectSuccess,comId:{},connectAndSendTask:{}, localPort:{}", com.getComId(), netAddr, NetUtil.getLocalPort(connect.channel()));
                connect.channel().attr(NettyConfigHelper.comEntityAttrKey).set(com);
                connect.channel().attr(NettyConfigHelper.isClient).set(Boolean.TRUE);
                connect.channel().attr(NettyConfigHelper.channelState).set(ChannelState.Registered);
                NettyConfigHelper.group.add(connect.channel());
//                terminalChannelService.addTerminalNetAddress(com, connect.channel());
                ProtocolType type = ProtocolType.getType(protocolId);
                connect.channel().attr(NettyConfigHelper.protocol).set(type);
                addConnectInfo(com.getComId());
                deviceConnectService.sendByProtocol(type, connectorTask, connect.channel());
                return;
            }
            if (future.isDone() && !future.isSuccess()) {
                log.info("ConnectFailed,com:({}):{},connectAndSendTask:{}", com.getComId(), com.getModuleSn(), netAddr);
                deviceConnectService.sendDeviceResultMessage(new ConnectTaskResultDto(ConnectorTaskResult.ExecuteCode.fail, connectorTask.getAtomicTaskNo(), null, CONNECTION_TIMED_OUT.toString(), com.getComId(), connectorTask.getFlag(), connectorTask.getTicketNumber()), com);
            }
        }));
        connect.syncUninterruptibly();
    }

    private synchronized void initProtocolBootstrap(Long protocolId) {
        if (bootstrapMap.containsKey(protocolId)) {
            return;
        }
        Bootstrap bootstrap = new Bootstrap();
        ChannelInitializer channelInitializer = ChannelInitializerFactory.getChannelInitializer(protocolId);
        bootstrap.group(clientEventLoop).channel(NioSocketChannel.class)
                .option(ChannelOption.TCP_NODELAY, Boolean.TRUE).
                option(ChannelOption.SO_KEEPALIVE, Boolean.TRUE).handler(channelInitializer);
        bootstrapMap.put(protocolId, bootstrap);
    }

    public Boolean isConnectedInOtherNode(Long comId) {
        String connectionKey = (String) redisUtils.get(RedisKeys.comEntityTcpLiveNode(comId));
        if (connectionKey == null) {
            return false;
        }

        try {
            String hostAddress = Inet4Address.getLocalHost().getHostAddress();
            if (hostAddress.equalsIgnoreCase(connectionKey)) {
                return false;
            }
        } catch (UnknownHostException e) {
            log.error("getLocalHostFailed", e);
        }
        log.debug("comIdCurrentNode:{}, comId:{}", connectionKey, comId);
        return true;
    }

    public void addConnectInfo(Long comId) {
        String hostAddress = "unknown";
        try {
            hostAddress = Inet4Address.getLocalHost().getHostAddress();
        } catch (UnknownHostException e) {
            log.error("unknown host");
        }
        log.info("Setting TCPConnectInfo:comId:{},node:{}", hostAddress);
        redisUtils.set(RedisKeys.comEntityTcpLiveNode(comId), hostAddress, 60 * 60);
    }

    public void channelInactive(Channel channel) {
        ComEntity com = channel.attr(NettyConfigHelper.comEntityAttrKey).get();
        if (com != null) {
            redisUtils.del(RedisKeys.comEntityTcpLiveNode(com.getComId()));
        }
    }

}
